package com.intretech.tj.mqtt;

import com.intretech.tj.global.parent.TriggerParent;
import com.intretech.tj.mqtt.Source.Source;
import com.intretech.tj.mqtt.entity.MessageInfo;
import com.intretech.tj.mqtt.pipeline.MessagePipeline;
import com.intretech.tj.mqtt.pipeline.TopicPipeline;
import com.intretech.tj.mqtt.trigger.CloseTrigger;
import com.intretech.tj.mqtt.trigger.LinkTrigger;
import com.intretech.tj.mqtt.trigger.MessageTrigger;
import com.intretech.tj.mqtt.trigger.TopicMessageTrigger;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.vertx.mutiny.mqtt.MqttEndpoint;
import io.vertx.mutiny.mqtt.MqttTopicSubscription;
import io.vertx.mutiny.mqtt.messages.*;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.inject.Inject;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;

import static com.intretech.tj.mqtt.OutLog.*;

@Slf4j
@ApplicationScoped
public class MQTTHandler extends Source {
    private final LinkTrigger linkTrigger;

    private final MessageTrigger messageTrigger;
    private final TopicMessageTrigger topicMessageTrigger;
    private final CloseTrigger closeTrigger;


    private final MessagePipeline messagePipeline;
    private final TopicPipeline topicPipeline;

    @Inject
    public MQTTHandler(LinkTrigger linkTrigger, MessageTrigger messageTrigger, TopicMessageTrigger topicMessageTrigger, CloseTrigger closeTrigger, MessagePipeline messagePipeline, TopicPipeline topicPipeline) {
        this.linkTrigger = linkTrigger;
        this.messageTrigger = messageTrigger;
        this.topicMessageTrigger = topicMessageTrigger;
        this.closeTrigger = closeTrigger;
        this.messagePipeline = messagePipeline;
        this.topicPipeline = topicPipeline;
    }

    /**
     * 客户端处理函数
     * <h2>链接与断开</h2>
     * 链接时添加到连接池<br/>
     * 断开时从连接池删除并清空其点阅池相关
     * <h2>订阅与退订</h2>
     * 支持客户端对主题的订阅与退订<br/>
     * 使用{@code Map<String, ArrayList<TopicInfo>> topicMap}作为存储<br/>
     * 存储所有订阅相关信息</br>
     * <h2>消息转发</h2>
     * <ul>当发布qos=0</ul>
     * <li>发布者-Publish(qos=0)->服务器</li>
     * <li>服务器-Publish(订阅qos)->订阅者</li>
     * <ul>当发布qos=1</ul>
     * <li>发布者-Publish(qos=1)->服务器</li>
     * <li>服务器-Publish(订阅qos)->订阅者</li>
     * <li>服务器-PublishAcknowledge收到->发布者</li>
     * <ul>当发布qos=2</ul>
     * <li>发布者-Publish(qos=2)->服务器</li>
     * <li>服务器-PublishReceived接受->发布者</li>
     * <li>发布者-PublishRelease释放->服务器</li>
     * <li>服务器-Publish(订阅qos)->订阅者</li>
     * <li>服务器-PublishComplete结束->发布者</li>
     *
     * @param endpoint 客户端
     */
    public void endpointHandler(MqttEndpoint endpoint, boolean ws) {

        // 订阅主题
        endpoint.subscribeHandler(subscribe -> subscribeHandler(endpoint, subscribe));

        // 退订主题
        endpoint.unsubscribeHandler(unsubscribe -> unsubscribeHandler(endpoint, unsubscribe));

        // 收到发布消息处理
        endpoint.publishHandler(message -> publishHandler(endpoint, message));

        // 接收Ack消息处理
        endpoint.publishAcknowledgeMessageHandler(ackMessage -> publishAcknowledgeMessageHandler(endpoint, ackMessage));

        // 接收Rec消息处理
        endpoint.publishReceivedMessageHandler(recMessage -> publishReceivedMessageHandler(endpoint, recMessage));

        // 接收Rel消息处理
        endpoint.publishReleaseMessageHandler(this::publishReleaseMessageHandler);

        // 断开处理 v3 ws不需要区分
        if (!ws) endpoint.disconnectHandler(() -> closeHandler(endpoint));
        // 断开处理 v5
        endpoint.closeHandler(() -> closeHandler(endpoint));

        // 接受远程客户端连接
        connectMessageHandler(endpoint);
    }

    /**
     * 订阅主题处理
     *
     * @param endpoint  链接
     * @param subscribe 订阅消息
     */
    public void subscribeHandler(MqttEndpoint endpoint, MqttSubscribeMessage subscribe) {
        List<MqttQoS> grantedQosLevels = new ArrayList<>();
        for (MqttTopicSubscription subscription : subscribe.topicSubscriptions()) {
            String topic = subscription.topicName();
            // 排除空主题
            if (topic.isEmpty())
                continue;
            subscribe(topic, subscription.qualityOfService(), endpoint.clientIdentifier());
            log.info(outLog.getMessage(Subscribe, endpoint.clientIdentifier(), topic));
            if (retainMessageMap.containsKey(topic))
                retain(endpoint.clientIdentifier(), getQos(subscription.qualityOfService(), retainMessageMap.get(topic).qos()), topic, retainMessageMap.get(topic).buffer());
            grantedQosLevels.add(subscription.qualityOfService());
        }
        // 确认订阅请求
        endpoint.subscribeAcknowledge(subscribe.messageId(), grantedQosLevels);
    }

    /**
     * 退订主题处理
     *
     * @param endpoint    链接
     * @param unsubscribe 退订消息
     */
    public void unsubscribeHandler(MqttEndpoint endpoint, MqttUnsubscribeMessage unsubscribe) {
        for (String topic : unsubscribe.topics()) {
            unsubscribe(topic, endpoint.clientIdentifier());
            log.info(outLog.getMessage(Unsubscribe, endpoint.clientIdentifier(), topic));
        }
        // 确认退订请求
        endpoint.unsubscribeAcknowledge(unsubscribe.messageId());
    }

    /**
     * 收到消息处理
     *
     * @param endpoint 链接
     * @param message  消息
     */
    public void publishHandler(MqttEndpoint endpoint, MqttPublishMessage message) {
        // 管道
        messagePipeline.pipeline(endpoint.clientIdentifier(), message);
        topicPipeline.pipeline(message.topicName(), message);

        // 触发
        messageTrigger.trigger(endpoint.clientIdentifier(), message);
        topicMessageTrigger.trigger(message.topicName(), new TopicMessageTrigger.IDAndMessage(endpoint.clientIdentifier(), message));

        log.info(outLog.getMessage(PublishHandler, message.messageId(), endpoint.clientIdentifier(), message.qosLevel().value(), message.payload().toString()));
        MessageInfo messageInfo = new MessageInfo(endpoint.clientIdentifier(), message);
        // 接收
        receive(messageInfo);

        // 处理发布者qos=2
        if (message.qosLevel().equals(MqttQoS.EXACTLY_ONCE)) {// 回复rec，等待rel释放
            endpoint.publishReceived(message.messageId());
            return;
        }

        // 转发
        transmit(message);
    }

    /**
     * 收到Ack消息处理<br/>
     * 向订阅者发送qos1消息时会收到Ack消息<br/>
     * 不需要回复订阅者
     *
     * @param ackMessage ack消息
     */
    public void publishAcknowledgeMessageHandler(MqttEndpoint endpoint, MqttPubAckMessage ackMessage) {
        log.info(outLog.getMessage(Acknowledge, ackMessage.messageId()));
        reply(endpoint.clientIdentifier(), ackMessage.messageId());
    }

    /**
     * 收到Rec消息处理<br/>
     * 向订阅者发送qos2消息时会收到Rec消息<br/>
     * 需要回复订阅者Rel消息
     *
     * @param endpoint   订阅者链接
     * @param recMessage rec消息
     */
    public void publishReceivedMessageHandler(MqttEndpoint endpoint, MqttPubRecMessage recMessage) {
        log.info(outLog.getMessage(Release, recMessage.messageId()));
        endpoint.publishRelease(recMessage.messageId());
        reply(endpoint.clientIdentifier(), recMessage.messageId());
    }

    /**
     * 收到Rel消息处理<br/>
     * 接收到发布者qos2确认的Rel消息<br/>
     * 接收到该消息开始进行向订阅者发布
     *
     * @param relMessage rel消息
     */
    public void publishReleaseMessageHandler(MqttPubRelMessage relMessage) {
        log.info(outLog.getMessage(Release, relMessage.messageId()));
        transmit(qosMessageMap.get(relMessage.messageId()).getMessage());
    }

    /**
     * 断开处理
     *
     * @param endpoint 链接
     */
    public void closeHandler(MqttEndpoint endpoint) {
        removeEndpoint(endpoint.clientIdentifier());
        unsubscribe(endpoint.clientIdentifier());
        closeTrigger.trigger(TriggerParent.ALL, endpoint.clientIdentifier());
        log.info(outLog.getMessage(Close, endpoint.clientIdentifier()));
    }

    /**
     * 链接处理
     *
     * @param endpoint 链接
     */
    public void connectMessageHandler(MqttEndpoint endpoint) {
        addEndpoint(endpoint);
        endpoint.accept();
        linkTrigger.trigger(endpoint.clientIdentifier(), endpoint);
        log.info(outLog.getMessage(Connect, endpoint.clientIdentifier()));
    }
}
